package com.serverless.framework.config;

import org.joda.time.format.DateTimeFormat;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.function.Consumer;

@Configuration
@Component
public class FunctionConfig {

    /**
     * 自动提交版本
     */
    @Bean
    public Consumer<String> printMessageAutoCommit() {
        return message -> {
            String currentTime = DateTimeFormat.fullTime().print(System.currentTimeMillis());
            System.out.println(currentTime + "Auto: " + message);
        };
    }

    /**
     * 手动提交版本
     *
     * @return Consumer<Message < String>>
     */
    @Bean
    public Consumer<Message<String>> printMessageManualCommit() {
        return message -> {
            String payload = message.getPayload();

            String currentTime = DateTimeFormat.fullTime().print(System.currentTimeMillis());
            System.out.println(currentTime + "Manual: " + payload);

            Acknowledgment acknowledgment = message.getHeaders()
                    .get("kafka_acknowledgment", Acknowledgment.class);

            if (acknowledgment != null) {
                acknowledgment.acknowledge();
            }
        };
    }

    /**
     * 批量处理版本
     *
     * @return Consumer<List < String>>
     */
    @Bean
    public Consumer<List<String>> printMessageBatch() {
        return messages -> {
            String currentTime = DateTimeFormat.fullTime().print(System.currentTimeMillis());
            System.out.println(currentTime + "Batch: " + String.join(";", messages));
        };
    }

}